package com.spring.example.rocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.function.Supplier;

/** 消息发送工具
 * author: xiaochi
 * date: 2025/9/30
 */
// MqProducerTemplate.java
@Component
@Slf4j
public class MqProducerTemplate {

    private final long timeOut = 3000;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通消息
     * @param message 消息体
     * @return
     * @param <T>
     */
    public <T> MessageSendResult send(MqMessage<T> message) {
        return doSend(message, null);
    }

    /**
     * 发送延迟消息
     * @param message 消息体
     * @param delaySeconds 延迟秒数
     * @return
     * @param <T>
     */
    public <T> MessageSendResult sendDelay(MqMessage<T> message, int delaySeconds) {
        message.setDelayLevel(delaySeconds);
        return doSend(message, () -> rocketMQTemplate.syncSend(message.getTopic(), buildMessage(message), timeOut, getDelayLevelBySeconds(delaySeconds)));
    }

    /**
     * 发送顺序消息
     * @param message  消息体
     * @param shardingKey 分区键
     * @return
     * @param <T>
     */
    public <T> MessageSendResult sendOrderly(MqMessage<T> message, String shardingKey) {
        return doSend(message, () -> rocketMQTemplate.syncSendOrderly(message.getTopic(), buildMessage(message), shardingKey));
    }

    /**
     * 发送事务消息
     * @param message  消息体
     * @param arg
     * @return
     * @param <T>
     */
    public <T> MessageSendResult sendTransaction(MqMessage<T> message, Object arg) {
        return doSend(message, () -> {
            //TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(message.getTopic(), msg, arg);
            //return SendResultConverter.convert(result);
            //return result;
            return rocketMQTemplate.sendMessageInTransaction(message.getTopic(), buildMessage(message), arg);
        });
    }

    /**
     * 异步发送
     */
    public <T> void sendAsync(MqMessage<T> message, SendCallback callback) {
        try {
            String destination = getDestination(message.getTopic(), message.getTag());
            rocketMQTemplate.asyncSend(destination, buildMessage(message), callback);
        } catch (Exception e) {
            log.error("异步发送消息失败, topic: {}, tag: {}", message.getTopic(), message.getTag(), e);
            //throw new MqException(message.getMessageId(), message.getTopic(), message.getTag(), "异步发送消息失败");
        }
    }

    /**
     * 执行发送
     * @param message 消息体
     * @param sendFunction 发送函数
     * @return
     * @param <T>
     */
    private <T> MessageSendResult doSend(MqMessage<T> message, Supplier<SendResult> sendFunction) {
        long startTime = System.currentTimeMillis();
        try {
            SendResult sendResult;
            if (sendFunction != null) {
                sendResult = sendFunction.get();
            } else {
                sendResult = rocketMQTemplate.syncSend(getDestination(message.getTopic(), message.getTag()), buildMessage(message));
            }
            log.info("消息发送成功, messageId: {}, topic: {}, tag: {}", message.getMessageId(), message.getTopic(), message.getTag());
            MessageSendResult result = MessageSendResult.success(sendResult.getMsgId());
            result.setCostTime(System.currentTimeMillis() - startTime);
            return result;

        } catch (Exception e) {
            log.error("消息发送失败, topic: {}, tag: {}, error: {}", message.getTopic(), message.getTag(), e.getMessage(), e);
            MessageSendResult result = MessageSendResult.failure(e.getMessage());
            result.setCostTime(System.currentTimeMillis() - startTime);
            return result;
        }
    }

    /**
     * 构建消息
     * @param message
     * @return
     * @param <T>
     */
    private <T>Message<MqMessage<T>> buildMessage(MqMessage<T> message) {
        return MessageBuilder.withPayload(message)
                .setHeader("businessKey", message.getBusinessKey())
                .setHeader(RocketMQHeaders.TAGS, message.getTag())
                .build();
        /*Message msg1 = new org.apache.rocketmq.common.message.Message();
        msg1.setTopic(message.getTopic());
        msg1.setTags(message.getTag());
        msg1.setBody(JSON.toJSONString(message.getBody()).getBytes());
        msg1.setKeys(message.getMessageId());
        // 设置延迟级别
        if (message.getDelayLevel() != null) {
            msg1.setDelayTimeLevel(message.getDelayLevel());
        }
        // 设置自定义属性
        msg1.putUserProperty("businessKey", message.getBusinessKey());
        msg1.putUserProperty("sendTime", String.valueOf(message.getSendTime().getTime()));
        msg1.putUserProperty("retryTimes", String.valueOf(message.getRetryTimes()));
        return msg1;*/
    }

    /**
     * 获取目标主题
     */
    private String getDestination(String topic, String tag) {
        return StringUtils.isBlank(tag) ? topic : topic + ":" + tag;
    }

    /**
     * 根据延迟秒数获取延迟级别
     */
    public static int getDelayLevelBySeconds(long seconds) {
        if (seconds <= 1) return 1;
        if (seconds <= 5) return 2;
        if (seconds <= 10) return 3;
        if (seconds <= 30) return 4;
        if (seconds <= 60) return 5;
        if (seconds <= 120) return 6;
        if (seconds <= 180) return 7;
        if (seconds <= 240) return 8;
        if (seconds <= 300) return 9;
        if (seconds <= 360) return 10;
        if (seconds <= 420) return 11;
        if (seconds <= 480) return 12;
        if (seconds <= 540) return 13;
        if (seconds <= 600) return 14;
        if (seconds <= 1200) return 15;
        if (seconds <= 1800) return 16;
        if (seconds <= 3600) return 17;
        return 18; // 超过1小时使用最大级别
    }

    /**
     * 根据延迟级别获取延迟秒数
     */
    public static long getSecondsByDelayLevel(int delayLevel) {
        switch (delayLevel) {
            case 1: return 1;
            case 2: return 5;
            case 3: return 10;
            case 4: return 30;
            case 5: return 60;
            case 6: return 120;
            case 7: return 180;
            case 8: return 240;
            case 9: return 300;
            case 10: return 360;
            case 11: return 420;
            case 12: return 480;
            case 13: return 540;
            case 14: return 600;
            case 15: return 1200;
            case 16: return 1800;
            case 17: return 3600;
            case 18: return 7200;
            default: return 0;
        }
    }
}
